
#include "DataTransmissionUdp.h"
#include "WingsTcpUdpProto.h"
#include "Proc.h"
#include "Conf.h"
#include "Misc.h"
#include "Crc.h"
#include "myLog.h"
#include "string.h"
// #include "errno.h"


using namespace std;

DataTransmissionUdp::~DataTransmissionUdp() 
{
    this->deinit();
}



Result DataTransmissionUdp::init(const initializer_list<void *> attr) 
{
    m_localPort = Conf::currentConf()->udp.port;
    m_headerSize = Conf::currentConf()->udp.headerSize;
    m_sliceSize = Conf::currentConf()->udp.chunkSize;
    m_type = Conf::currentConf()->typeValue;

    m_sock = ::socket(AF_INET, SOCK_DGRAM, 0);
    if (m_sock < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::InitFailed);
        res.appendInfo("udp data transmission, create udp socket failed");
        res.appendInfo(info);

        return res;
    }

    auto exitClosure = [&](const Result &res){
        if (::close(this->m_sock) < 0) 
        {
            int err = errno;
            qlog_e("udp data transmission, close udp socket failed, detail: %s", getErrorInfo(err).c_str());
        }
        this->m_sock = -1;

        return res;
    };

    // set recv timeout.
    timeval timeout = { 0 };
    timeout.tv_sec = 3;
    timeout.tv_usec = 0;
    auto ret = setsockopt(m_sock, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeval));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("udp data transmission, set udp socket recv timeout failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    int bufSize = 256 * 1024;
    ret = setsockopt(m_sock, SOL_SOCKET, SO_RCVBUF, &bufSize, sizeof(int));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("udp data transmission, set udp socket recv buffer size failed");
        res.appendInfo(info);

        return exitClosure(res);
    }

    // bind
    sockaddr_in localAddr = { 0 };
    localAddr.sin_family = AF_INET;
    localAddr.sin_addr.s_addr = INADDR_ANY;
    localAddr.sin_port = htons(m_localPort);
    ret = ::bind(m_sock, reinterpret_cast<sockaddr *>(&localAddr), sizeof(sockaddr_in));
    if (ret < 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("udp data transmission, bind udp socket failed");
        res.appendInfo(info);

        return exitClosure(res);
    }
    m_peerAddrSize = sizeof(m_peerAddr);
    m_workerExitFlag = false;
    m_worker = thread(&DataTransmissionUdp::run, this);
    setThreadName("udp background");

    return Result();
}

Result DataTransmissionUdp::transmit(const void *src, const size_t size) 
{
    if (nullptr == src || 0 == size) 
    {
        auto res = Result::create(ErrorKind::InvalidArgument);
        res.appendInfo("udp data transmission, passed invalid arguments to transmit function");

        return res;
    }

    auto ret = ::sendto(m_sock, src, size, 0, reinterpret_cast<sockaddr *>(&m_peerAddr), m_peerAddrSize);
    if (ret <= 0) 
    {
        int err = errno;
        auto info = getErrorInfo(err);
        auto res = Result::create(ErrorKind::ExecuteFailed);
        res.appendInfo("udp data transmission, transmit failed, detail: ");
        res.appendInfo(info);

        return res;
    }
    
    if (size != static_cast<size_t>(ret)) 
    {
        return Result::create(ErrorKind::ImcompleteIo, "udp data transmission, sent bytes is not equal to data size");
    }

    return Result();
}

void DataTransmissionUdp::release(const void *item) 
{
    if (nullptr == item) 
    {
        return;
    }
    auto ptr = const_cast<WingsInternalWrapper *>(reinterpret_cast<const WingsInternalWrapper *>(item));
    delete ptr;
}

void DataTransmissionUdp::deinit()  
{
    if (false == m_workerExitFlag) 
    {
        m_workerExitFlag = true;

        if (true == m_worker.joinable()) 
        {
            m_worker.join();
        }
        else 
        {
            m_worker.detach();
        }
    }

    if (::close(this->m_sock) < 0) 
    {
        int err = errno;
        qlog_e("udp data transmission, close udp socket failed(deinit), detail: %s", getErrorInfo(err).c_str());
    }
    this->m_sock = -1;

}

void DataTransmissionUdp::run() 
{
    qlog_i("wings udp thread start.");
    vector<uint8_t> buffer;
    uint8_t buf[1024];
    int ret = -1;
    const uint16_t header = 0xaa55, tail = 0x0022;
    uint32_t bodySize = 0, nameSize = 0, remain = 0;
    uint32_t recvSize = 0;
    uint32_t crc32 = 0;
    uint32_t tmp32 = 0;
    uint16_t tmp16 = 0;

    // 当接收数据时，每次收到一帧，都需要给对端回复确认信息，表明本次接收到了多少字节，如果UDP出现了丢包，对端可以立即感知到。当接收完毕，校验数据包的时候，也需要回复对端，表明这次收到的完整数据包是否合法。
    // 如多对端接收超时，就应该认为是本次传输失败，应该重新发送一个完整的数据包。超时时间至少设置为100ms.
    // 回复字符串消息格式: 
    // 接收:
    // 状态字符串 + ',' + 接收到的字节数, 当接收到的字节数是负数字符串时，表示本次接收失败,-1表示收到的数据包语义有错误(无效的数据)，-2表示接收失败，其他值是未定义的。
    // 举例: "header,10"
    //       "body,1280"
    // 处理:
    // 状态字符串 + ',' + 状态码, 状态码是0表示处理完毕，-1表示数据域语义有错误(无效的数据)，-2表示crc校验错误，其他值是未定义的。
    // 举例: "process,0"
    auto response = [&](const char *stat, const int value){
        string msg = stat;
        msg += ",";
        msg += to_string(value);
        msg += "\n";

        auto res = this->transmit(msg.data(), msg.size());
        if (false == res.isOk()) 
        {
            qlog_e("response failed, %s", res.m_message.c_str());
            return false;
        }

        return true;
    };
    
    while (true) 
    {
        if (true == m_workerExitFlag) 
        {
            break;
        }

        // recv header.
        buffer.resize(m_headerSize, 0);
        ret = ::recvfrom(m_sock, buffer.data(), buffer.size(), 0, reinterpret_cast<sockaddr *>(&m_peerAddr),  (socklen_t *)&m_peerAddrSize);
        if (ret <= 0) 
        {
            int err = errno;
            if (EWOULDBLOCK != err && EINTR != err) 
            {
                qlog_e("udp data transmission, recv header failed, detail: %s", getErrorInfo(err).c_str());
                // response("header", -2);
            }

            continue;
        }

        if (m_headerSize != ret) 
        {
            qlog_e("udp data transmission, header size is %d bytes but recv %d bytes", m_headerSize, ret);
            response("header", ret);
            continue;
        }

        tmp16 = static_cast<uint16_t>(buffer[1]) << 8 | static_cast<uint16_t>(buffer[0]);
        if (WINGS_HEADER != tmp16) 
        {
            qlog_e("udp data transmission, invalid header.");
            response("header", -1);
            continue;
        }

        tmp32 = static_cast<uint32_t>(buffer[7]) << 24 | static_cast<uint32_t>(buffer[6]) << 16 | static_cast<uint32_t>(buffer[5]) << 8 | static_cast<uint32_t>(buffer[4]);
        if (m_type != (uint16_t)tmp32) 
        {
            qlog_e("udp data transmission, invalid type '%d', it's not for me.", *(reinterpret_cast<decltype(header) *>(buffer.data()) + 1));
            response("header", -1);
            continue;
        }
        //获取包体数据长度
        bodySize = static_cast<uint32_t>(buffer[11]) << 24 | static_cast<uint32_t>(buffer[10]) << 16 | static_cast<uint32_t>(buffer[9]) << 8 | static_cast<uint32_t>(buffer[8]);
        tmp32 = static_cast<uint32_t>(buffer[15]) << 24 | static_cast<uint32_t>(buffer[14]) << 16 | static_cast<uint32_t>(buffer[13]) << 8 | static_cast<uint32_t>(buffer[12]);
        if (WINGS_TAIL != tmp32) 
        {
            qlog_e("udp data transmission, invalid tail '%d', it's not for me.", *(reinterpret_cast<decltype(header) *>(buffer.data()) + 1));
            response("header", -1);
            continue;
        }
        
        qlog_v("Success recv %d bytes header from '%s:%d'.", ret, inet_ntoa(m_peerAddr.sin_addr), ntohs(m_peerAddr.sin_port));

        if(bodySize > 0)
        {
            // recv remaining.
            buffer.reserve(m_headerSize + bodySize);
            remain = bodySize;
            recvSize = m_sliceSize;
            //接收剩余数据
            while(true) 
            {
                if (remain <= 0) 
                {
                    response("body size", bodySize);
                    break;
                }

                if (0 == remain / m_sliceSize) 
                {
                    recvSize = remain % m_sliceSize;
                }
                vector<uint8_t> slice(recvSize, 0);
                ret = ::recvfrom(m_sock, slice.data(), slice.size(), 0, reinterpret_cast<sockaddr *>(&m_peerAddr),  (socklen_t *)&m_peerAddrSize);
                if (ret <= 0) 
                {
                    int err = errno;
                    // if (EWOULDBLOCK != err) 
                    // {
                        qlog_e("udp data transmission, recv body failed, detail: %s", getErrorInfo(err).c_str());
                    // }
                    break;
                }
                buffer.insert(buffer.end(), slice.begin(), slice.begin() + ret);
                remain -= ret;
            }
        }
        
        auto wrapper = new(WingsInternalWrapper);
        auto reply = this->dataResolver(move(buffer),*wrapper);
        if(reply != nullptr) {
            int size = reinterpret_cast<WingsUpdateHeaderTcpUdp *>(reply)->dataSize + m_headerSize;
            this->transmit(reply,size);
        }
        if(!wrapper->name.empty()) {
            this->asyncRecv(wrapper, sizeof(WingsInternalWrapper));
            response("process", 0);
        }
        
    }

    qlog_i("wings udp thread exit.");
}

